package ar.edu.itba.pod.legajo45129.event;

import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.legajo45129.agentSimulation.DistributedMultithreadEventDispatcher;
import ar.edu.itba.pod.legajo45129.manager.NodeManager;
import ar.edu.itba.pod.multithread.EventDispatcher;

import com.google.common.base.Preconditions;

public class RemoteEventDispatcherImpl implements RemoteEventDispatcher {
	public static final long POLLING_WAIT_TIME = 5000;
	public static final long REMOVE_WAIT = 10000;
	private static Logger logger = Logger
			.getLogger(RemoteEventDispatcherImpl.class);
	private BlockingQueue<EventInformation> events = new LinkedBlockingQueue<EventInformation>();
	private ConcurrentMap<EventInformation, Long> processedEvents  = new ConcurrentHashMap<EventInformation, Long>();
	private ConcurrentHashMap<NodeInformation, Long> lastNodeUpdates = new ConcurrentHashMap<NodeInformation, Long>();


	private EventDispatcher eventDispatcher;
	private PublishEvents publishEventsThread;
	private PollingEventsForMe pollingEventsThread;
	private RemoveProcessedEvent removeProcessedEvent;

	public RemoteEventDispatcherImpl(EventDispatcher eventDispatcher)
			throws RemoteException {
		Preconditions.checkNotNull(eventDispatcher);
		UnicastRemoteObject.exportObject(this, 0);
		NodeInformation localNode = NodeManager.getInstance()
				.getNodeInformation();
		Registry registry;
		registry = LocateRegistry.getRegistry(localNode.host(),
				localNode.port());
		if (registry == null) {
			registry = LocateRegistry.createRegistry(localNode.port());
		}

		logger.debug("Binding " + Node.DISTRIBUTED_EVENT_DISPATCHER);
		registry.rebind(Node.DISTRIBUTED_EVENT_DISPATCHER, this);
		this.eventDispatcher = eventDispatcher;
		this.publishEventsThread = new PublishEvents();
		this.publishEventsThread.start();
		NodeManager.getInstance().addThread(this.publishEventsThread);
		this.pollingEventsThread = new PollingEventsForMe();
		this.pollingEventsThread.start();
		NodeManager.getInstance().addThread(this.pollingEventsThread);
		this.removeProcessedEvent = new RemoveProcessedEvent();
		this.removeProcessedEvent.start();
		NodeManager.getInstance().addThread(this.removeProcessedEvent);
	}

	@Override
	public boolean publish(EventInformation event) throws RemoteException,
	InterruptedException {
		if (events.contains(event)) {
			return false;
		}
		events.add(event);
		return true;

	}

	@Override
	public Set<EventInformation> newEventsFor(NodeInformation nodeInformation)
			throws RemoteException {
		Set<EventInformation> result = new HashSet<EventInformation>();
		Iterator<EventInformation> it;
		synchronized (processedEvents) {
			it = new CopyOnWriteArrayList<EventInformation>(processedEvents.keySet()).iterator();
		}
		Long processedTime, updateTime;

		while (it.hasNext()) {
			EventInformation e = it.next();
			processedTime = processedEvents.get(e);
			updateTime = lastNodeUpdates.get(nodeInformation);

			if (updateTime != null && updateTime < processedTime) {
				result.add(e);
				lastNodeUpdates.put(nodeInformation, System.currentTimeMillis());
			}
		}
		return result;
	}

	@Override
	public BlockingQueue<Object> moveQueueFor(Agent agent)
			throws RemoteException {
		return ((DistributedMultithreadEventDispatcher) eventDispatcher)
				.deregister(agent);
	}

	private class PublishEvents extends Thread {
		@Override
		public void run() {
			//			Random random = new Random();

			while (true) {
				try {
					EventInformation eventInformation = events.take();

					// publish to connected nodes

					// if event is from another node, then publish in this node
					if (!eventInformation.nodeId().equals(NodeManager.getInstance().getNodeInformation().id())) {
						((DistributedMultithreadEventDispatcher) eventDispatcher).localPublish(eventInformation.source(), eventInformation.event());
					}

					Long now = System.currentTimeMillis();

					// sync?
					processedEvents.put(eventInformation, now);
					Set<NodeInformation> connectedNodes;
					try {
						connectedNodes = NodeManager.getInstance().getClusterAdministration().connectedNodes();
						for (NodeInformation n: connectedNodes) {
							NodeManager.getInstance().getRemoteEventDispatcher(n.host(), n.port()).publish(eventInformation);
							lastNodeUpdates.put(n, now);
						}
					} catch (RemoteException e) {
						logger.warn("could not connect connected node");
					} catch (NotBoundException e) {
						logger.warn("could not connect to connected node");
					}
					logger.trace(eventInformation + " catched from queue");
					((DistributedMultithreadEventDispatcher) eventDispatcher)
					.localPublish(eventInformation.source(),
							eventInformation.event());
				} catch (InterruptedException e) {
					logger.warn("Polling events to put into local event dispatcher Interrupted");
				}
			}
		}
	}

	private class PollingEventsForMe extends Thread {
		@Override
		public void run() {
			NodeInformation me = NodeManager.getInstance().getNodeInformation();
			try {
				while (true) {
					sleep(POLLING_WAIT_TIME);

					Set<NodeInformation> connectedNodes;
					connectedNodes = NodeManager.getInstance()
							.getClusterAdministration().connectedNodes();
					for (NodeInformation ni : connectedNodes) {
						RemoteEventDispatcher red = null;
						try {
							red = NodeManager.getInstance()
									.getRemoteEventDispatcher(ni.host(),
											ni.port());
							Set<EventInformation> newEventsForMe = red
									.newEventsFor(me);
							if (newEventsForMe != null
									&& newEventsForMe.size() > 0) {
								synchronized (events) {
									events.addAll(newEventsForMe);
								}
							}
						} catch (NotBoundException e) {
							logger.warn("could not reach " + ni + ": " + e);
						}
					}
				}
			} catch (RemoteException e) {
				logger.warn("could not get connected nodes");
			} catch (InterruptedException e) {
				logger.warn("polling interrupted");
			}
		}
	}

	private class RemoveProcessedEvent extends Thread {
		private static final long EPS = 5000;

		@Override
		public void run() {
			try {
				while (true) {
					sleep(REMOVE_WAIT);

					Iterator<EventInformation> it;
					synchronized (processedEvents) {
						it = new CopyOnWriteArrayList<EventInformation>(processedEvents.keySet()).iterator();
					}
					
					while (it.hasNext()) {
						EventInformation e = it.next();
						if ((System.currentTimeMillis() - processedEvents.get(e)) > EPS) {
							processedEvents.remove(e);
						}
					}
				}
			} catch (InterruptedException e) {
				logger.warn("Remove processed events interrupted");
			}
		}
	}

}
